package pgcluster

import (
	"phant-operator/pkg/apis/phant/v1alpha1"
	"phant-operator/pkg/apis/phant/v1alpha1/defaulting"
	"phant-operator/pkg/apis/phant/v1alpha1/validation"
	"phant-operator/pkg/controller"
	"phant-operator/pkg/manager"
	"phant-operator/pkg/metrics"

	v1 "k8s.io/api/core/v1"
	apiequality "k8s.io/apimachinery/pkg/api/equality"
	errorutils "k8s.io/apimachinery/pkg/util/errors"
	"k8s.io/client-go/tools/record"
	"k8s.io/klog/v2"
)

// ControlInterface implements the control logic for updating TidbClusters and their children StatefulSets.
// It is implemented as an interface to allow for extensions that provide different semantics.
// Currently, there is only one implementation.
type ControlInterface interface {
	// UpdateTidbCluster implements the control logic for StatefulSet creation, update, and deletion
	UpdatePostgresCluster(*v1alpha1.PostgresCluster) error
}

type defaultPostgresClusterControl struct {
	pgControl controller.PostgresClusterControlInterface

	patroniManager       manager.Manager
	patroniMemberManager manager.Manager

	postgresMemberManager manager.Manager

	recorder record.EventRecorder
}

// NewDefaultTidbClusterControl returns a new instance of the default implementation TidbClusterControlInterface that
// implements the documented semantics for TidbClusters.
func NewDefaultPostgresClusterControl(
	patroniManager manager.Manager,
	patroniMemberManager manager.Manager,
	postgresMemberManager manager.Manager,
	recorder record.EventRecorder) ControlInterface {
	return &defaultPostgresClusterControl{
		patroniManager:        patroniManager,
		patroniMemberManager:  patroniMemberManager,
		postgresMemberManager: postgresMemberManager,
		recorder:              recorder,
	}
}

// UpdateTracker executes track the deleted postgrescluster CR and lagacy CR
func (c *defaultPostgresClusterControl) UpdateTracker(pg *v1alpha1.PostgresCluster) error {
	//c.patroniMemberManager.Sync()
	return nil
}

// UpdatePostgresCluster executes the core logic loop for a PostgresCluster.
func (c *defaultPostgresClusterControl) UpdatePostgresCluster(pg *v1alpha1.PostgresCluster) error {
	c.defaulting(pg)
	if !c.validate(pg) {
		return nil // fatal error, no need to retry on invalid object
	}

	var errs []error
	oldStatus := pg.Status.DeepCopy()

	//to-do: postgres cluster without patroni
	if pg.Spec.Patroni == nil {
		klog.Errorln("No partroni configured, skip to launch postgres cluster")
		return nil // fatal error, no need to retry on invalid object

	}

	if err := c.updatePatroniCluster(pg); err != nil {
		errs = append(errs, err)
	}

	if apiequality.Semantic.DeepEqual(&pg.Status, oldStatus) {
		return errorutils.NewAggregate(errs)
	}

	return errorutils.NewAggregate(errs)
}

func (c *defaultPostgresClusterControl) updatePatroniCluster(pg *v1alpha1.PostgresCluster) error {
	c.recordMetrics(pg)

	ns := pg.GetNamespace()
	pgName := pg.GetName()
	klog.Infof("start to update postgresql cluster %s/%s.", ns, pgName)

	// to-do: add reconcile logic
	// Add patroniManager to create SerivceAccount with Sync method
	if err := c.patroniManager.Sync(pg); err != nil {
		metrics.ClusterUpdateErrors.WithLabelValues(ns, pgName, "patroni").Inc()
		return err
	}
	// Add ConfigMap for PostgresCluster user password logic
	if err := c.postgresMemberManager.Sync(pg); err != nil {
		metrics.ClusterUpdateErrors.WithLabelValues(ns, pgName, "postgres").Inc()
		return err
	}

	if err := c.patroniMemberManager.Sync(pg); err != nil {
		metrics.ClusterUpdateErrors.WithLabelValues(ns, pgName, "patroni").Inc()
		return err
	}
	return nil
}

func (c *defaultPostgresClusterControl) defaulting(pg *v1alpha1.PostgresCluster) {
	defaulting.SetPostgresClusterDefault(pg)
}

func (c *defaultPostgresClusterControl) validate(pg *v1alpha1.PostgresCluster) bool {
	errs := validation.ValidatePostgresCluster(pg)
	if len(errs) > 0 {
		aggregatedErr := errs.ToAggregate()
		klog.Errorf("postgresql cluster %s/%s is not valid and must be fixed first, aggregated error: %v", pg.GetNamespace(), pg.GetName(), aggregatedErr)
		c.recorder.Event(pg, v1.EventTypeWarning, "FailedValidation", aggregatedErr.Error())
		return false
	}
	return true
}

func (c *defaultPostgresClusterControl) recordMetrics(pg *v1alpha1.PostgresCluster) {
	ns := pg.GetNamespace()
	pgName := pg.GetName()
	if pg.Spec.Postgres != nil {
		metrics.ClusterSpecReplicas.WithLabelValues(ns, pgName, "postgres").Set(float64(pg.Spec.Postgres.Replicas))
	}

}
